package com.xrui.hbase.schema;

import com.google.common.base.Preconditions;
import com.google.common.base.Objects;

import com.google.common.collect.*;

import com.xrui.hbase.AvroSchemaResolver;
import com.xrui.hbase.ColumnName;
import com.xrui.hbase.codec.RowKeyDecoder;
import com.xrui.hbase.codec.RowKeyEncoder;
import com.xrui.hbase.codec.impl.RawBytesRowKeyDecoder;
import com.xrui.hbase.codec.impl.RawBytesRowKeyEncoder;
import com.xrui.hbase.conversion.CellConverter;
import com.xrui.hbase.exception.InvalidLayoutException;
import com.xrui.hbase.exception.NoSuchColumnException;
import com.xrui.hbase.schema.desc.CellSchemaDesc;
import com.xrui.hbase.schema.desc.ColumnDesc;
import com.xrui.hbase.schema.desc.TableLayoutDesc;
import com.xrui.hbase.schema.impl.AvroCellSchema;
import com.xrui.hbase.schema.impl.CustomCellSchema;
import com.xrui.hbase.schema.impl.GenericCellSchema;
import com.xrui.hbase.schema.impl.ProtobufCellSchema;
import com.xrui.hbase.schema.spec.CellSpec;
import com.xrui.hbase.util.NameValidator;
import org.apache.avro.Schema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;


/**
 * Layout of a HBase table.
 * <p>
 * <p>
 * TableLayout wraps a layout descriptor represented as a
 * {@link TableLayoutDesc TableLayoutDesc} Avro record.
 * TableLayout provides strict validation and accessors to navigate through the layout.
 * <p>
 * <p>
 * KijiTableLayouts can be created via one of two methods: from a concrete layout with
 * {@link #newLayout(TableLayoutDesc)}.
 * For the format requirements of layout descriptors for these methods, see the
 * "Layout descriptors" section below.
 * <p>
 * <h1>Overall structure</h1>
 * <p>At the top-level, a table contains:
 * <ul>
 * <li>the table name and description;</li>
 * <li>how row keys are encoded;</li>
 * <li>the table locality groups.</li>
 * </ul>
 * <p>
 * <p>Each locality group has:
 * <ul>
 * <li>a primary name, unique within the table, a description and some name aliases;</li>
 * <li>whether the data is to be stored in memory or on disk;</li>
 * <li>data retention lifetime;</li>
 * <li>maximum number of versions to keep;</li>
 * <li>type of compression;</li>
 * <li>column families stored in this locality group</li>
 * </ul>
 * <p>
 * <p>Each column family has:
 * <ul>
 * <li>a primary name, globally unique within the table,
 * a description and some name aliases;</li>
 * <li>for map-type families, the Avro schema of the cell values;</li>
 * <li>for group-type families, the collection of columns in the group.</li>
 * </ul>
 * <p>
 * <p>Each column in a group-type family has:
 * <ul>
 * <li>a primary name, unique within the family, a description and some name aliases;</li>
 * <li>an Avro schema.</li>
 * </ul>
 * <p>
 * <h1>Layout descriptors</h1>
 * <p>
 * Layout descriptors are represented using
 * {@link TableLayoutDesc TableLayoutDesc} Avro records.
 * Layout descriptors come in two flavors: <i>concrete layouts</i> and <i>layout updates</i>.
 * <p>
 * <h2><i>Concrete layout descriptors</i></h2>
 * A concrete layout descriptors is an absolute, standalone description of a table layout, which
 * does not reference or build upon any previous version of the table layout. Column IDs have
 * been assigned to all locality groups, families and columns.
 * <p>
 * <p> Names of tables, locality groups, families and column qualifiers must be valid identifiers.
 * Name validation occurs in {@link NameValidator NameValidator}.
 * <p>
 * <h3>Validation rules</h3>
 * <p>
 * <ul>
 * <li> Table names, locality group names, family names, and column names in a group-type family
 * must be valid identifiers (no punctuation or symbols).
 * Note: map-type family qualifiers are free-form, but do never appear in a table layout.
 * <li> Locality group names and aliases must be unique within the table.
 * <li> Family names and aliases must be unique within the table.
 * <li> Group-type family qualifiers must be unique within the family.
 * </ul>
 * <p>
 * <h2><i>Layout update descriptors</i></h2>
 * A table layout update descriptor builds on a reference table layout, and describes layout
 * modification to resolve on the reference layout.
 * This mechanism prevents race conditions when updating the layout of a table.
 * The first layout of a newly created table has no reference layout.
 * <p>
 * <p>During a layout update, the user may delete or declare new locality groups, families and/or
 * columns, or modify existing entities, by specifying the new layout.  Update validation rules
 * are enforced to ensure compatibility (see Validation rules for updates below).
 * <p>
 * <p>Entities may also be renamed, as long as uniqueness requirements are met.
 * Primary name updates must be explicitly annotated by setting the {@code renamedFrom} field of
 * the entity being renamed.
 * The name of a table cannot be changed.
 * <p>
 * <p>For example, suppose the reference layout contained one family {@code Info}, containing a
 * column {@code Name}, and the user wishes to add a new {@code Address} column to the
 * {@code Info} family.
 * To perform this update, the user would create a layout update by starting with the existing
 * layout, setting the {@code reference_layout} field to the {@code layout_id} of the
 * current layout, and adding a new {@link ColumnDesc} record describing the {@code Address}
 * column to the the {@code columns} field of the {@link ColumnDesc} for the {@code Info} family.
 * <p>
 * <p>The result of applying a layout update on top of a concrete reference layout is a new
 * concrete layout.
 * <p>
 * <h3> Validation rules for updates </h3>
 * <p>
 * <p> Updates are subject to the same restrictions as concrete layout descriptors.
 * In addition:</p>
 * <p>
 * <ul>
 * <li> The type of a family (map-type or group-type) cannot be changed.
 * <li> A family cannot be moved into a different locality group.
 * <li> The encoding of HBase cells (hash, UID, final) cannot be modified.
 * <li> The schema of a HBase cell can only be changed to a schema that is compatible with
 * all the former schemas of the column. Schema compatibility requires that the new schema
 * allows decoding all former schemas associated to the column or the map-type family.
 * </ul>
 * <p>
 * <h1>Row keys encoding</h1>
 * <p>
 * A row in a HBase table is identified by its HBase row key. HBase row keys are converted into HBase
 * row keys according to the row key encoding specified in the table layout:
 * <ul>
 * <li> Raw encoding: the user has direct control over the encoding of row keys in the HBase
 * table. In other words, the HBase row key is exactly the HBase row key. These are used
 * when the user would like to use arrays of bytes as row keys.
 * </li>
 * <li> Formatted: the row key is comprised of one or more components. Each component can be
 * a string, a number or a hash of another component. The user will specify the size
 * of this hash. The user also specifies the actual order of the components in the key.
 * </li>
 * </ul>
 * <p>
 * Hashing allows to spread the rows evenly across all the regions in the table. Specifying the size
 * of the hash gives the user fine grained control of how the data will be distributed.
 * <p>
 * <h1>Cell schema</h1>
 * <p>
 * HBase cells are encoded according to a schema specified via
 * {@link CellSchema CellSchema}.
 * <p>
 */
public final class TableLayout {
    private static final Logger LOG = LoggerFactory.getLogger(TableLayout.class);
    /**
     * Avro record describing the table layout absolutely (no reference layout required).
     */
    private final TableLayoutDesc mDesc;
    /**
     * Row key in the table.
     */
    private final RowKeyFormat mRowKeyFormat;
    private final RowKeyEncoder mRowKeyEncoder;
    private final RowKeyDecoder mRowKeyDecoder;
    /**
     * Columns in the table, in no particular order.
     */
    private final ImmutableList<ColumnLayout> mColumnLayouts;
    /**
     * Columns in the table, in no particular order.
     */
    private final ImmutableSet<ColumnName> mColumns;
    /**
     * Map family names and aliases to family layout.
     */
    private final ImmutableMap<String, ColumnLayout> mColumnMap;
    /**
     * Optional schema resolver that allows resolution of Avro schemas.
     */
    private AvroSchemaResolver mSchemaResolver;

    /**
     * Constructs a TableLayout from an Avro descriptor and an optional reference layout.
     *
     * @param desc Avro layout descriptor (relative to the reference layout).
     * @throws InvalidLayoutException if the descriptor is invalid or inconsistent wrt reference.
     */
    private TableLayout(TableLayoutDesc desc)
        throws InvalidLayoutException {
        // Deep-copy the descriptor to prevent mutating a parameter:
        //mDesc = TableLayoutDesc.newBuilder(Preconditions.checkNotNull(desc)).build();
        mDesc = desc;

        if (!NameValidator.isValidTableName(getName())) {
            throw new InvalidLayoutException(String.format("Invalid table name: '%s'.", getName()));
        }

        isValidRowKeyFormat(mDesc.getKeysFormat());
        mRowKeyFormat = mDesc.getKeysFormat();
        switch (mRowKeyFormat.getEncoding()) {
            case FORMATTED:
                mRowKeyDecoder = RowKeyDecoder.Factory.create(mRowKeyFormat.getDecoder());
                mRowKeyEncoder = RowKeyEncoder.Factory.create(mRowKeyFormat.getEncoder());
                break;
            case RAW:
                mRowKeyDecoder = RowKeyDecoder.Factory.create(RawBytesRowKeyDecoder.PROVIDER_NAME);
                mRowKeyEncoder = RowKeyEncoder.Factory.create(RawBytesRowKeyEncoder.PROVIDER_NAME);
                break;
            default:
                throw new RuntimeException("Unsupported row key encoding " + mRowKeyFormat.getEncoding());
        }


        /* All the columns in the table. */
        final List<ColumnLayout> columns = Lists.newArrayList();
        /* Map from column name to column layout. */
        final Map<String, ColumnLayout> columnMap = Maps.newHashMap();
        /* All primary column names (including map-type families). */
        final Set<ColumnName> columnNames = Sets.newTreeSet();

        for (ColumnDesc columnDesc : desc.getColumns()) {
            final ColumnName column = ColumnName.create(columnDesc.getFamily(), columnDesc.getQualifier());
            ColumnLayout columnLayout = new ColumnLayout(columnDesc);
            validateCellSchema(columnLayout.getSchema());
            if (null != columnMap.put(column.getName(), columnLayout)) {
                throw new InvalidLayoutException(String.format(
                    "Layout for table '%s' contains duplicate column '%s'.",
                    getName(), column));
            }

            columns.add(columnLayout);
            columnNames.add(column);
        }


        if (columns.isEmpty()) {
            throw new InvalidLayoutException("Missing descriptor(s) for column(s)");
        }

        mColumnLayouts = ImmutableList.copyOf(columns);
        mColumnMap = ImmutableMap.copyOf(columnMap);
        mColumns = ImmutableSet.copyOf(columnNames);
    }

    /**
     * Validates a cell schema descriptor.
     * <p>
     * Ignores failures due to specific Avro record classes not being present on the classpath.
     *
     * @param schema New cell schema descriptor.
     * @throws InvalidLayoutException if the cell schema descriptor is invalid
     *                                or incompatible with the reference cell schema.
     */
    private static void validateCellSchema(CellSchema schema)
        throws InvalidLayoutException {
        CellSpec.readAvroSchema(schema);
    }

    /**
     * Creates and returns a new TableLayout instance as specified by an Avro TableLayoutDesc
     * description record.
     *
     * @param layout The Avro TableLayoutDesc descriptor record that describes the actual layout.
     *               of the table.
     * @return A new table layout.
     * @throws InvalidLayoutException If the descriptor is invalid.
     */
    public static TableLayout newLayout(TableLayoutDesc layout) throws InvalidLayoutException {
        return new TableLayout(layout);
    }

    /**
     * Ensure a row key format specified in a layout file is sane.
     *
     * @param format The RowKeyFormat created from the layout file for a table.
     * @throws InvalidLayoutException If the format is invalid.
     */
    private void isValidRowKeyFormat(RowKeyFormat format) throws InvalidLayoutException {
        // For RAW encoding, ignore the rest of the fields.
        if (format.getEncoding() == RowKeyEncoding.RAW) {
            return;
        }

        // At least one primitive component.
        if (format.getComponents().size() <= 0) {
            throw new InvalidLayoutException("At least 1 component is required in row key format.");
        }


        Set<String> nameSet = new HashSet<>();
        for (RowKeyComponent component : format.getComponents()) {
            // ensure names are valid "[a-zA-Z_][a-zA-Z0-9_]*"
            if (!NameValidator.isValidName(component.getName())) {
                throw new InvalidLayoutException("Names should begin with a letter followed by a "
                    + "combination of letters, numbers and underscores.");
            }
            nameSet.add(component.getName());
        }

        // repeated component names
        if (nameSet.size() != format.getComponents().size()) {
            throw new InvalidLayoutException("Component name already used.");
        }
    }

    public Schema getAvroSchema(String schemaID) {
        return Preconditions.checkNotNull(mSchemaResolver).apply(schemaID);
    }

    /**
     * Returns the Avro descriptor for this table layout.
     *
     * @return the Avro descriptor for this table layout.
     */
    public TableLayoutDesc getDesc() {
        return mDesc;
    }

    /**
     * Returns the table name.
     *
     * @return the table name.
     */
    public String getName() {
        return mDesc.getName();
    }

    /**
     * Returns the map from column name to layouts.
     *
     * @return the map from column name to layouts.
     */
    public Map<String, ColumnLayout> getColumnMap() {
        return mColumnMap;
    }

    /**
     * Returns all the column layouts in the table.
     *
     * @return all the column layouts in the table.
     */
    public Collection<ColumnLayout> getColumnLayouts() {
        return mColumnLayouts;
    }

    /**
     * Returns all the columns in the table, in no particular order.
     *
     * @return all the columns in the table, in no particular order.
     */
    public Set<ColumnName> getColumns() {
        return mColumns;
    }

    /**
     * Reports the raw specification record for the specified column.
     * <p>
     * <p> Note: in most cases, you should use {@link #getCellSpec(ColumnName)}. </p>
     *
     * @param columnName Column to reports the raw specification record of.
     * @return the raw specification record for the specified column.
     * @throws NoSuchColumnException if the column does not exist.
     */
    public CellSchema getCellSchema(ColumnName columnName) throws NoSuchColumnException {
        final ColumnLayout columnLayout = mColumnMap.get(columnName.getName());
        if (columnLayout == null) {
            if (columnName.isFullyQualified()) {
                // There is no cell schema for the specified fully-qualified column.
                // Try the family (this will only work for map-type families):
                return getCellSchema(ColumnName.create(columnName.getFamily(), null));
            } else {
                throw new NoSuchColumnException(String.format(
                    "Table '%s' has no column '%s'.", getName(), columnName.getName()));
            }
        }
        return columnLayout.mSchema;
    }

    /**
     * Reports the record converter for the specified column.
     * <p>
     * <p> Note: in most cases, you should use {@link #getCellSpec(ColumnName)}. </p>
     *
     * @param columnName Column to reports the raw specification record of.
     * @return the converter of record for the specified column.
     * @throws NoSuchColumnException if the column does not exist.
     */
    public CellConverter getConverter(ColumnName columnName) throws NoSuchColumnException {
        final ColumnLayout columnLayout = mColumnMap.get(columnName.getName());
        if (columnLayout == null) {
            if (columnName.isFullyQualified()) {
                // There is no cell converter for the specified fully-qualified column.
                // Try the family (this will only work for map-type families):
                return getConverter(ColumnName.create(columnName.getFamily(), null));
            } else {
                throw new NoSuchColumnException(String.format(
                    "Table '%s' has no column '%s'.", getName(), columnName.getName()));
            }
        }
        String converter = columnLayout.getSchema().getConverter();
        if(null != converter) {
            return CellConverter.Factory.create(converter);
        } else {
            return null;
        }
    }

    // -----------------------------------------------------------------------------------------------

    /**
     * Reports the specification of the specified column.
     *
     * @param column Column to report the specification of.
     * @return the specification for the specified column.
     * @throws IOException on I/O error.
     */
    public CellSpec getCellSpec(ColumnName column) throws IOException {
        return CellSpec.fromCellSchema(getCellSchema(column))
            .setSchemaResolver(mSchemaResolver);
    }

    /**
     * Reports whether a column exists.
     *
     * @param column Column name.
     * @return whether the specified column exists.
     */
    public boolean exists(ColumnName column) {
        final ColumnLayout columnLayout = mColumnMap.get(column.getName());
        return columnLayout != null;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean equals(Object other) {
        if (!(other instanceof TableLayout)) {
            return false;
        }
        final TableLayout otherLayout = (TableLayout) other;
        return getDesc().equals(otherLayout.getDesc());
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode() {
        return getDesc().hashCode();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        return Objects.toStringHelper(TableLayout.class)
            .add("Row Key", mRowKeyFormat)
            .add("Cell Schema", mColumnLayouts)
            .toString();
    }

    /**
     * Returns the schema table bound to this table layout. Null means unbound.
     *
     * @return the schema table bound to this table layout. Null means unbound.
     */
    public AvroSchemaResolver getSchemaResolver() {
        return mSchemaResolver;
    }

    public RowKeyFormat getRowKeyFormat() {
        return mRowKeyFormat;
    }

    /**
     * Binds this table layout to the specified schema table.
     * <p>
     * <p> Once set, the schema table bound to a table layout cannot be modified. </p>
     *
     * @param schemaResolver Avro schema resolver to bind this layout to.
     * @return this layout.
     */
    public TableLayout setSchemaResolver(AvroSchemaResolver schemaResolver) {
        Preconditions.checkState(mSchemaResolver == null);
        mSchemaResolver = schemaResolver;
        return this;
    }

    public RowKeyEncoder getRowKeyEncoder() {
        return mRowKeyEncoder;
    }

    public RowKeyDecoder getRowKeyDecoder() {
        return mRowKeyDecoder;
    }

    /**
     * Concrete layout of a column.
     */
    public final class ColumnLayout {

        /**
         * Family layout descriptor.
         */
        private final ColumnDesc mDesc;
        /**
         * Column name.
         */
        private final ColumnName mName;
        /**
         * Cell schema.
         */
        private final CellSchema mSchema;

        /**
         * Builds a new column layout instance.
         *
         * @param columnDesc Descriptor of the column.
         * @throws InvalidLayoutException if the layout is invalid or inconsistent.
         */
        public ColumnLayout(ColumnDesc columnDesc)
            throws InvalidLayoutException {
            mDesc = Preconditions.checkNotNull(columnDesc);
            mName = ColumnName.create(columnDesc.getFamily(), columnDesc.getQualifier());
            CellSchemaDesc schemaDesc = mDesc.getCellSchema();
            switch (schemaDesc.getType()) {
                case AVRO:
                    mSchema = AvroCellSchema.create(schemaDesc.getId(), schemaDesc.getSchema(),
                        schemaDesc.getCompression(), schemaDesc.getConverter());
                    break;
                case PROTOBUF:
                    mSchema = ProtobufCellSchema.create(schemaDesc.getId(), schemaDesc.getSchema(),
                        schemaDesc.getCompression(), schemaDesc.getConverter());
                    break;
                case RAW_BYTES:
                    mSchema = GenericCellSchema.create(SchemaType.RAW_BYTES, schemaDesc.getSchema(),
                        schemaDesc.getCompression(), schemaDesc.getConverter());
                    break;
                case CDFSTORE:
                    mSchema = GenericCellSchema.create(SchemaType.CDFSTORE, schemaDesc.getSchema(),
                            schemaDesc.getCompression(), schemaDesc.getConverter());
                    break;
                case CUSTOM:
                    mSchema = CustomCellSchema.create(schemaDesc.getId(), schemaDesc.getSchema(),
                        schemaDesc.getCompression(), schemaDesc.getConverter());
                    break;
                default:
                    throw new RuntimeException("Unhandled cell encoding: " + schemaDesc.getType());
            }

            if (!NameValidator.isValidName(columnDesc.getFamily())) {
                throw new InvalidLayoutException(String.format(
                    "Invalid family name: '%s'.", columnDesc.getFamily()));
            }
        }

        /**
         * @return the Avro descriptor for this family.
         */
        public ColumnDesc getDesc() {
            return mDesc;
        }

        /**
         * @return the name for the column.
         */
        public ColumnName getName() {
            return mName;
        }

        public CellSchema getSchema() {
            return mSchema;
        }

        @Override
        public String toString() {
            return Objects.toStringHelper(RowKeyFormat.class)
                .add("name", mName)
                .add("schema", mSchema)
                .toString();
        }
    }

}
